package com.mjf.chain

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala._

/**
 * 默认的共享组为default，slotSharingGroup会使后续的组都改变。不同组之间不能使用同一个slot
 */
object SlotSharingGroupDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(8) // 批模式单并行度本地调试也会导致程序卡死不动
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)  // 流模式本地调试会导致程序卡死不动

    val source: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

    val result: DataStream[String] = source
      .filter(_.>(5))
      .slotSharingGroup("filter") // 设置操作的槽共享组。Flink 会将具有相同 slot 共享组的操作放在同一个 slot 中
      .map(_ + "大于5")
      .slotSharingGroup("map")
      .map(_ + "!!!")

    result.print()

    env.execute(SlotSharingGroupDemo.getClass.getName)

  }
}
